package exporter;


import com.google.common.base.Joiner;
import org.apache.commons.collections.map.HashedMap;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.text.DecimalFormat;
import java.util.*;

/**
 * Created by xbkaishui on 16/10/23.
 */
public class Exporter {

	public static void main(String[] args) throws Exception {

		String hbaseTable = "tbdp:H_Product_Fund";
		if (args.length > 0) {
			hbaseTable = args[0];
		}
		System.out.println("hbase table " + hbaseTable);
		String type = args[1];
		if (StringUtils.equalsIgnoreCase(type, "1")) {
			System.out.println("exportMainData");
			exportMainData(hbaseTable);
		}
		if (StringUtils.equalsIgnoreCase(type, "2")) {
			System.out.println("exportVolRate");
			exportVolRate(hbaseTable);
		}
		if (StringUtils.equalsIgnoreCase(type, "3")) {
			System.out.println("exportKlineData");
			exportKlineData(hbaseTable);
		}
		if (StringUtils.equalsIgnoreCase(type, "4")) {
			System.out.println("deleteData");
			deleteData(hbaseTable);
//			testDelete(hbaseTable);
		}
	}

	private static void testDelete(String hbaseTable)  throws  Exception{
		HTableInterface hTable = Utils.getHbaseTable(hbaseTable);
//			String rowKey = generateRowKey("308000_2_20160906");
			Delete delete = new Delete(Bytes.toBytes("308000_2_20160906"));
			hTable.delete(delete);
	}

	private static void deleteData(String hbaseTable) throws Exception {
		Connection sqlServerConn = Utils.getConnection();

		String hisSql = "SELECT\n"
				+ "\tCode AS scode,\n"
				+ "\tMarket AS market,\n"
				+ "\tt_Date AS tdate\n"
				+ "\t\tFROM\n"
				+ "\t\t\tmintMapping.dbo.g_TopViewDataSplit_His";

		String dboSql = "SELECT\n"
				+ "\tcode AS code,\n"
				+ "\tmarket+1 AS market,\n"
				+ "\ttdate AS tdate\n"
				+ "FROM\n"
				+ "\tmintMapping.dbo.JG_K_SDL  WHERE\n"
				+ "\ttdate < '20161107'";

		hisSql = Utils.readFile("./sql/del_his.sql");
		System.out.println(hisSql);

		dboSql = Utils.readFile("./sql/del_dbo.sql");
		System.out.println(dboSql);

		Statement statement = sqlServerConn.createStatement();
		ResultSet rs = statement.executeQuery(dboSql);
		Set<String> lines = new HashSet<>();
		while (rs.next()) {
			String scode = rs.getString(1);
			String marketType = rs.getString(2);
			String tDate = rs.getString(3);
			String line = Joiner.on(",").join(scode, marketType, tDate);
			lines.add(line);
		}


		rs = statement.executeQuery(hisSql);
		while (rs.next()) {
			String scode = rs.getString(1);
			String marketType = rs.getString(2);
			String tDate = rs.getString(3);
			String line = Joiner.on(",").join(scode, marketType, tDate);
			lines.remove(line);
		}

		System.out.println("total delete records: " + lines.size());
		for (String line : lines) {
			String rowKey = generateRowKey(line);
			System.out.println(line + "rowkey: " + rowKey);
		}


		HTableInterface hTable = Utils.getHbaseTable(hbaseTable);
		List<Delete> deletes = new ArrayList<>();
		for (String line : lines) {
			String rowKey = generateRowKey(line);
			Delete delete = new Delete(Bytes.toBytes(rowKey));
			deletes.add(delete);
			if(deletes.size()==500){
				hTable.delete(deletes);
				deletes.clear();
			}
		}
		if(deletes.size()>0){
			hTable.delete(deletes);
		}
	}

	private static String generateRowKey(String line) {
		String[] cols = line.split(",");
		String scode = cols[0];
		String marketType = cols[1];
		String tDate = cols[2];
		String rowkey = StringUtils.reverse(scode) + "_" + marketType + "_" + tDate;
		return rowkey;
	}

	private static void exportVolRate(String hbaseTable) throws Exception {
		Connection sqlServerConn = Utils.getConnection();
		//send vol rate
		String volrateSql = "SELECT \n" + "\tcode as scode,\n" + "\tmarket + 1 AS market,\n"
				+ "\ttdate as tdate,\n" + "\tsdl_1d3w_buy*100 as agencybvolrate,\n"
				+ "\tsdl_1d3w_allbuy*100 as allbvolrate,\n" + "\tsdl_1d3w_sell*100 as agencysvolrate,\n"
				+ "\tsdl_1d3w_allsell*100 as allsvolrate,\n" + "\tsdl_1d3w*100 as agency1bsvolrate,\n"
				+ "\tsdl_3d3w*100 as agency3bsvolrate,\n" + "\tsdl_5d3w*100 as agency5bsvolrate,\n"
				+ "\tsdl_10d3w*100 as agency10bsvolrate,\n" + "\tsdl_20d3w*100 as agency20bsvolrate,\n"
				+ "\tzhang_fu*100 as allvolrate\n" + "FROM\n" + "\tmintMapping.dbo.JG_K_SDL\n"
				+ "WHERE\n" + " tdate < '20160215' and tdate > '20160101' order by tdate asc ";

		volrateSql = Utils.readFile("./sql/rate.sql");
		System.out.println(volrateSql);

		HTableInterface hTable = Utils.getHbaseTable(hbaseTable);

		Statement statement = sqlServerConn.createStatement();
		Map<String, List<Map<String, Object>>> dataList = new LinkedHashMap<>();

		statement = sqlServerConn.createStatement();
		ResultSet rs = statement.executeQuery(volrateSql);
		while (rs.next()) {
			Map<String, Object> curRow = new HashedMap();
			String SCode = rs.getString(1);
			String MarketType = rs.getString(2);
			String TDate = rs.getString(3);
			curRow.put("SCode", SCode);
			curRow.put("MarketType", MarketType);
			curRow.put("TDate", TDate);

			Double AgencyBVolRate = rs.getDouble(4);
			Double AllBVolRate = rs.getDouble(5);
			Double AgencySVolRate = rs.getDouble(6);
			Double AllSVolRate = rs.getDouble(7);
			Double Agency1BSVolRate = rs.getDouble(8);
			Double Agency3BSVolRate = rs.getDouble(9);
			Double Agency5BSVolRate = rs.getDouble(10);
			Double Agency10BSVolRate = rs.getDouble(11);
			Double Agency20BSVolRate = rs.getDouble(12);
			Double AllVolRate = rs.getDouble(13);

			curRow.put("AgencyBVolRate", AgencyBVolRate);
			curRow.put("AllBVolRate", AllBVolRate);
			curRow.put("AgencySVolRate", AgencySVolRate);
			curRow.put("AllSVolRate", AllSVolRate);
			curRow.put("Agency1BSVolRate", Agency1BSVolRate);
			curRow.put("Agency3BSVolRate", Agency3BSVolRate);
			curRow.put("Agency5BSVolRate", Agency5BSVolRate);
			curRow.put("Agency10BSVolRate", Agency10BSVolRate);
			curRow.put("Agency20BSVolRate", Agency20BSVolRate);
			curRow.put("AllVolRate", AllVolRate);

			String key = MarketType + "_" + SCode;
			List<Map<String, Object>> stockList = dataList.get(key);
			if (stockList == null) {
				stockList = new LinkedList<>();
				dataList.put(key, stockList);
			}

			stockList.add(curRow);
		}
		sendOut(dataList, hbaseTable);
	}

	/**
	 * 导出 kline数据到ProundFund
	 *
	 * @param hbaseTable
	 */
	private static void exportKlineData(String hbaseTable) throws Exception {
		Connection sqlServerConn = Utils.getConnection();
		String klineSql =
				"SELECT\n" + "  Code AS scode,\n" + "  Market AS markettype,\n" + "  T_Date AS tdate,\n"
						+ "  AClose AS atrade,\n" + "  AOpen AS aopen,\n" + "  AHigh AS ahigh,\n"
						+ "  ALow AS alow,\n" + "  AClose AS aclose,\n" + "  YClose AS yclose,\n"
						+ "  Vol AS vol,\n" + "  Amt AS amt,\n" + "  1 AS istrade,\n" + "  CASE\n"
						+ "WHEN(Code LIKE '60%' AND Market = 1)\n" + "OR(Code LIKE '30%' AND Market = 2)\n"
						+ "OR(Code LIKE '00%' AND Market = 2) THEN\n" + "  0\n" + "ELSE\n" + "  1\n"
						+ "END AS isexp,\n" + " CASE\n" + "WHEN(Code LIKE '60%' AND Market = 1)\n"
						+ "OR(Code LIKE '30%' AND Market = 2)\n"
						+ "OR(Code LIKE '00%' AND Market = 2) THEN\n" + "  12\n"
						+ "WHEN(Code LIKE '00%' AND Market = 1)\n"
						+ "OR(Code LIKE '39%' AND Market = 2) THEN\n" + "  11\n" + "ELSE\n" + "  13\n"
						+ "END AS codetyep\n" + "FROM\n" + "  Fbase.dbo.StockHis\n" + "WHERE\n"
						+ "  T_Date < '20160215' and  T_Date > '20160101' ";

		klineSql = Utils.readFile("./sql/kline.sql");
		System.out.println(klineSql);

		HTableInterface hTable = Utils.getHbaseTable(hbaseTable);
		Statement statement = sqlServerConn.createStatement();
		ResultSet rs = statement.executeQuery(klineSql);
		List<Map<String, Object>> rows = new ArrayList<>(100);
		while (rs.next()) {
			Map<String, Object> curRow = new HashedMap();
			String SCode = rs.getString(1);
			String MarketType = rs.getString(2);
			String TDate = rs.getString(3);
			String ATrade = rs.getString(4);
			String AOpen = rs.getString(5);
			String AHigh = rs.getString(6);
			String ALow = rs.getString(7);
			String AClose = rs.getString(8);
			String YClose = rs.getString(9);
			String Vol = rs.getString(10);
			String Amt = rs.getString(11);
			String IsTrade = rs.getString(12);
			String IsExp = rs.getString(13);
			String CodeType = rs.getString(14);

			curRow.put("SCode", SCode);
			curRow.put("MarketType", MarketType);
			curRow.put("TDate", TDate);
			curRow.put("ATrade", ATrade);
			curRow.put("AOpen", AOpen);
			curRow.put("AHigh", AHigh);
			curRow.put("ALow", ALow);
			curRow.put("AClose", AClose);
			curRow.put("YClose", YClose);
			curRow.put("Vol", Vol);
			curRow.put("Amt", Amt);
			curRow.put("IsTrade", IsTrade);
			curRow.put("IsExp", IsExp);
			curRow.put("CodeType", CodeType);

			rows.add(curRow);

			if (rows.size() == 99) {
				sendRows(rows, hTable);
				rows.clear();
			}
		}

		if (!rows.isEmpty()) {
			sendRows(rows, hTable);
		}
	}

	private static void exportMainData(String hbaseTable) throws SQLException, IOException {
		Connection sqlServerConn = Utils.getConnection();

		//使用map获取数据 key 是market_SCODE  value 是股票列表
		Map<String, List<Map<String, Object>>> dataList = new LinkedHashMap<>();

		String amtSoSql =
				"SELECT  Code, Market, t_Date, AgencyBuy * 10000, PublicBuy * 10000, AgencySell * 10000, PublicSell * 10000, "
						+ "                AgencyBuySco, AgencySellSco, PriTeamBuySco, PriTeamSellSco, PublicBuySco, "
						+ "                PublicSellSco, AllSco FROM mintMapping.dbo.g_TopViewDataSplit_His "
						+ "                WHERE t_Date >= '20160101' AND t_Date < '20160215' order by t_Date asc ";


		amtSoSql = Utils.readFile("./sql/agency.sql");
		System.out.println(amtSoSql);

		Statement statement = sqlServerConn.createStatement();
		ResultSet rs = statement.executeQuery(amtSoSql);
		while (rs.next()) {
			Map<String, Object> curRow = new HashedMap();
			String SCode = rs.getString(1);
			String MarketType = rs.getString(2);
			String TDate = rs.getString(3);

			Double AgencyBAmt = rs.getDouble(4);
			Double AllBAmt = rs.getDouble(5);
			Double AgencySAmt = rs.getDouble(6);
			Double AllSAmt = rs.getDouble(7);

			curRow.put("SCode", SCode);
			curRow.put("MarketType", MarketType);

			curRow.put("TDate", TDate);
			curRow.put("AgencyBAmt", AgencyBAmt);
			curRow.put("AllBAmt", AllBAmt);
			curRow.put("AgencySAmt", AgencySAmt);
			curRow.put("AllSAmt", AllSAmt);

			Integer AgencyBSco = rs.getInt(8);
			Integer AgencySSco = rs.getInt(9);
			Integer PriTeamBSco = rs.getInt(10);
			Integer PriTeamSSco = rs.getInt(11);
			Integer PublicBSco = rs.getInt(12);
			Integer PublicSSco = rs.getInt(13);
			Integer AllSco = rs.getInt(14);

			curRow.put("AgencyBSco", AgencyBSco);
			curRow.put("AgencySSco", AgencySSco);
			curRow.put("PriTeamBSco", PriTeamBSco);
			curRow.put("PriTeamSSco", PriTeamSSco);
			curRow.put("PublicBSco", PublicBSco);
			curRow.put("PublicSSco", PublicSSco);
			curRow.put("AllSco", AllSco);



			String key = MarketType + "_" + SCode;

			List<Map<String, Object>> stockList = dataList.get(key);
			if (stockList == null) {
				stockList = new LinkedList<>();
				dataList.put(key, stockList);
			}

			//计算 5日平均资金综合分值
			int start = stockList.size() - 1;
			int end = Math.max(start - 4, 0);
			int sumSco = AllSco;
			int cnt = 1;
			for (int i = start; i > end; i--) {
				//                System.out.println(cnt);
				cnt++;
				sumSco += (Integer) stockList.get(i).get("AllSco");
			}

			Integer Avg5AllSco = sumSco / cnt;
			curRow.put("Avg5AllSco", Avg5AllSco);

			//top资金流向
			Double All1BSAmt = AllBAmt - AllSAmt;
			Double All3BSAmt = 0.0;
			Double All5BSAmt = 0.0;
			Double All10BSAmt = 0.0;
			//计算 主力增减周期
			Double Agency1BSAmt = AgencyBAmt - AgencySAmt;
			Double Agency3BSAmt = 0.0;

			curRow.put("All1BSAmt", All1BSAmt);
			curRow.put("Agency1BSAmt", Agency1BSAmt);

			int total = stockList.size();
			if (total >= 2) {
				List<Map<String, Object>> sub = stockList.subList(total - 2, total);
				for (Map<String, Object> row : sub) {
					Agency3BSAmt += (Double) row.get("AgencyBAmt") - (Double) row.get("AgencySAmt");
					All3BSAmt += (Double) row.get("AllBAmt") - (Double) row.get("AllSAmt");
				}
				Agency3BSAmt += Agency1BSAmt;
				All3BSAmt += All1BSAmt;
			}

			curRow.put("Agency3BSAmt", Agency3BSAmt);
			curRow.put("All3BSAmt", All3BSAmt);

			Double Agency5BSAmt = 0.0;
			if (total >= 4) {
				List<Map<String, Object>> sub = stockList.subList(total - 4, total);
				for (Map<String, Object> row : sub) {
					Agency5BSAmt += (Double) row.get("AgencyBAmt") - (Double) row.get("AgencySAmt");
					All5BSAmt += (Double) row.get("AllBAmt") - (Double) row.get("AllSAmt");
				}
				All5BSAmt += All1BSAmt;
			}
			curRow.put("Agency5BSAmt", Agency5BSAmt);
			curRow.put("All5BSAmt", All5BSAmt);

			Double Agency7BSAmt = 0.0;
			if (total >= 6) {
				List<Map<String, Object>> sub = stockList.subList(total - 6, total);
				for (Map<String, Object> row : sub) {
					Agency7BSAmt += (Double) row.get("AgencyBAmt") - (Double) row.get("AgencySAmt");
				}
				Agency7BSAmt += Agency1BSAmt;
			}
			curRow.put("Agency7BSAmt", Agency7BSAmt);

			if (total >= 9) {
				List<Map<String, Object>> sub = stockList.subList(total - 9, total);
				for (Map<String, Object> row : sub) {
					All10BSAmt += (Double) row.get("AllBAmt") - (Double) row.get("AllSAmt");
				}
				All10BSAmt += All1BSAmt;
			}
			curRow.put("All10BSAmt", All10BSAmt);
			String TTime = "15:00";
			curRow.put("TTime", TTime);
			stockList.add(curRow);
		}
		sendOut(dataList, hbaseTable);

	}

	private static void sendOut(Map<String, List<Map<String, Object>>> dataList, String hbaseTable)
			throws IOException {
		//        String hbase_product_fund = "tbdp:H_Product_Fund";
		HTableInterface table = Utils.getHbaseTable(hbaseTable);
		Iterator<String> keyIterator = dataList.keySet().iterator();
		Random rand = new Random();
		while (keyIterator.hasNext()) {
			String key = keyIterator.next();
			System.out.println("start write key " + key);
			List<Map<String, Object>> rows = dataList.get(key);
			sendRows(rows, table);
			try {
				Thread.sleep(10 + rand.nextInt(50));
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}

	}

	private static void sendRows(List<Map<String, Object>> rows, HTableInterface table)
			throws IOException {
		List<Put> puts = new ArrayList<>(rows.size());
		for (Map<String, Object> data : rows) {
			String scode = (String) data.get("SCode");
			String marketType = (String) data.get("MarketType");
			String tDate = (String) data.get("TDate");
			String rowkey = StringUtils.reverse(scode) + "_" + marketType + "_" + tDate;
			Put row = new Put(Bytes.toBytes(rowkey));
			Iterator<String> keyIterator = data.keySet().iterator();
			while (keyIterator.hasNext()) {
				String key = keyIterator.next();
				Object val = data.get(key);
				String value = objToVal(val);
				row.addColumn(Bytes.toBytes("cf1"), Bytes.toBytes(key), Bytes.toBytes(value));
			}
			puts.add(row);
		}
		for (int i = 0; i < 3; i++) {
			try {
				table.put(puts);
				return;
			} catch (Throwable e) {
				e.printStackTrace();
				try {
					Thread.sleep(2 * 60000);
				} catch (InterruptedException e1) {
					e1.printStackTrace();
				}
			}
		}
	}

	/**
	 * send one row to hbase
	 *
	 * @param data
	 */
	private static void sendRow(Map<String, Object> data, HTableInterface hbaseTable)
			throws IOException {

		String scode = (String) data.get("SCode");
		String marketType = (String) data.get("MarketType");
		String tDate = (String) data.get("TDate");
		String rowkey = StringUtils.reverse(scode) + "_" + marketType + "_" + tDate;
		Put row = new Put(Bytes.toBytes(rowkey));
		Iterator<String> keyIterator = data.keySet().iterator();
		while (keyIterator.hasNext()) {
			String key = keyIterator.next();
			Object val = data.get(key);
			String value = objToVal(val);
			row.addColumn(Bytes.toBytes("cf1"), Bytes.toBytes(key), Bytes.toBytes(value));
		}
		hbaseTable.put(row);
	}

	private static String objToVal(Object val) {
		if (val instanceof Double) {
			DecimalFormat df = new DecimalFormat("###.##");
			return df.format(val);
		}
		return String.valueOf(val);
	}
}
